package com.hadwinling.alogriithm.projectforpso.mypso

import com.hadwinling.alogriithm.projectforpso.PSO_RDD.PSO_RDD.no_of_iteration_External
import com.hadwinling.alogriithm.projectforpso.PSO_RDD.Particle
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

import scala.collection.mutable.ArrayBuffer

object PSO_RDD extends Serializable {
  var dimension = 2 // 设置维度为2 ，即两个参数
  var no_of_particles = 1000 // 粒子数
  var no_of_iteration_External = 10    //找出10个最佳的全局最优
  var no_of_iteration_Internal = 1000 //设置迭代次数
  var c1 = 2  //加速系数，传统为2
  var c2 = 2  //加速系数，传统为2
  var w = 0.5 //官良权重，一般初始化为0.9，然后随着进化过程现行减到0.4
  var g_best=Array.fill(dimension)(math.random) //将 位置用math.random 填满2个


  def main(args:Array[String]): Unit ={
    val conf=new SparkConf().setAppName("mypso")
      .setMaster("local")

    val sc = new SparkContext(conf)

    var swarm = ArrayBuffer[Particle]() // 生成Particle类型的数组
    for(i <- 0 to no_of_particles-1){
      swarm += new Particle(dimension,i) with Serializable   //初始化Particle，并添加到数组中
    }
    var swarm_rdd = sc.parallelize(swarm,4)
    // 创建rdd，并设置8个分区（根据核心数量进行分区，我这台机子是4个核心），

    swarm_rdd.foreach(f => init_pbest(f))       // 设置当前的位置
    swarm_rdd.foreach(f => f.compute_pbest_fitness())  // 计算各粒子的适应度
    swarm_rdd.foreach(f => f.compute_pfitness())  //当前适应度的值
    swarm_rdd.foreach(f => global_best_position(f))  //更新粒子的位置

    var temp_rdd: RDD[Particle]= sc.emptyRDD[Particle] //创建一个空的rdd
    temp_rdd=swarm_rdd
    var updated_velocity_swarm: RDD[Particle]= sc.emptyRDD[Particle]

    var bCast = sc.broadcast(g_best)   //广播当前的最佳的粒子个体

    println(obj_func(g_best))
    for(iteration <- 0 to no_of_iteration_External-1){
      updated_velocity_swarm = temp_rdd.map{x => update_particle(x)}  //更新每个粒子的速度
      temp_rdd = updated_velocity_swarm
      temp_rdd.collect().foreach(f => global_best_position(f))   //计算新的全局最佳粒子
      println(iteration+"  "+obj_func(g_best))
      bCast.unpersist(blocking=true)   //从工作节点中去除就的全局最佳粒子
      bCast = sc.broadcast(g_best)//更新广播当前的最佳的粒子个体
    }

    /**
     *  更新粒子的速度和位置
     * @param p 粒子
     * @return
     */
    def update_particle(p:Particle):Particle={
      g_best = bCast.value            // 初始化广播全局最佳粒子
      for(iter <- 0 to no_of_iteration_Internal){        // 更新粒子操作
        for(i <- 0 to dimension-1){                         //更新粒子的速度操作
          var toward_pbest= c1*math.random*(p.p_best(i)-p.p_position(i))
          var toward_gbest =  c2*math.random*(g_best(i)-p.p_position(i))
          p.p_velocity(i) = w*p.p_velocity(i) + toward_pbest + toward_gbest
        }
        for(i <- 0 to dimension-1){   //更新速度
          p.p_position(i) = p.p_position(i) + p.p_velocity(i)
        }

        p.compute_pfitness()          //计算

        if(p.p_fitness < p.pbest_fitness){  //计算
          p.p_best = p.p_position
        }

        p.compute_pbest_fitness()      //computr p_best fitness

        if(p.pbest_fitness < obj_func(g_best)){      //compute gbest
          g_best=p.p_position
        }
      }
      return p  //return updated particle
    }
    sc.stop()
  }


  /**
   * 适应度函数
   * @param x
   * @return
   */
  def obj_func(x:Array[Double]):Double = {   //Sphere Function
    var temp:Double  =0
    for(dim <- 0 to x.length-1 )
    {
      temp =temp + (math.pow(x(dim),2))
    }
    return temp
  }

  /**
   * 更新位置
   * @param p
   */
  def global_best_position(p:Particle)={
    if(p.pbest_fitness < obj_func(g_best)){
      g_best=p.p_best
    }
  }

  /**
   * 初始化为
   * @param p
   */
  def init_pbest(p:Particle)={
    p.p_best = p.p_position
  }

}
